package com.mytest.provider.rll;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;
import redis.clients.jedis.JedisCommands;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * 代码中分为集群模式和单机模式，并且两者的方法、参数都一样，原因是spring封装的执行脚本的方法中
 * （ RedisConnection 接口继承于 RedisScriptingCommands 接口的 eval 方法），
 * 集群模式的方法直接抛出了不支持执行脚本的异常（虽然实际是支持的），
 * 所以只能拿到 Redis 的connection来执行脚本，而 JedisCluster 和 Jedis 中的方法又没有实现共同的接口，
 * 所以只能分开调用。
 * @Author: llq
 * @Date: 2020/5/18
 */
public class RedisDistributedLock extends AbstractDistributedLock {

    private final Logger logger = LoggerFactory.getLogger(RedisDistributedLock.class);

    private RedisTemplate<Object, Object> redisTemplate;

    private ThreadLocal<String> lockFlag = new ThreadLocal<String>();

    public static final String UNLOCK_LUA;

    /*@Autowired
    RedisTemplate redisTemplate;*/

    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
        sb.append("then ");
        sb.append("    return redis.call(\"del\",KEYS[1]) ");
        sb.append("else ");
        sb.append("    return 0 ");
        sb.append("end ");
        UNLOCK_LUA = sb.toString();
    }

    public RedisDistributedLock(RedisTemplate<Object, Object> redisTemplate) {
        super();
        this.redisTemplate = redisTemplate;
    }

    /**
     * lock
     * @param key
     * @param expire  过期时间
     * @param retryTimes 重试次数
     * @param sleepMillis 休眠时间
     * @return
     */
    @Override
    public boolean lock(String key, long expire, int retryTimes, long sleepMillis) {
        boolean result = setRedis(key, expire);
        // 如果获取锁失败，按照传入的重试次数进行重试
        while((!result) && retryTimes-- > 0){
            try {
                logger.debug("lock failed, retrying..." + retryTimes);
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                return false;
            }
            result = setRedis(key, expire);
        }
        return result;
    }

    private boolean setRedis(String key, long expire) {
        try {

            /**
             * 将锁资源放入 Redis （注意是当key不存在时才能放成功，所以使用 setIfAbsent 方法）：
             * redisTemplate.opsForValue().setIfAbsent("key", "value");
             * 设置过期时间
             * redisTemplate.expire("key", 30000, TimeUnit.MILLISECONDS);
             * 释放锁
             * redisTemplate.delete("key");
             */
            //setIfAbsent 方法之后线程挂掉了，即没有给锁定的资源设置过期时间，默认是永不过期，
            // 那么这个锁就会一直存在。所以需要保证设置锁及其过期时间两个操作的原子性
            // 保证设置 Redis 值和过期时间的原子性，避免前面提到的两次 Redis 操作期间出现意外而导致的锁不能释放的问题
            /*String result = (String) redisTemplate.execute(new RedisCallback<String>() {
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException {
                    *//*JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                    String uuid = UUID.randomUUID().toString();
                    lockFlag.set(uuid);
                    return commands.set(key, uuid, "NX", "PX", expire);*//*


                    StringRedisSerializer serializer = new StringRedisSerializer();
                    String uuid = UUID.randomUUID().toString();
                    lockFlag.set(uuid);
                    //仅仅key不存在时，则设置key的值为value,返回true,否则返回value
                    Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(uuid));
                    connection.close();
                    return success.toString();
                }
            });*/


            String uuid = UUID.randomUUID().toString();
            lockFlag.set(uuid);
            Boolean result = redisTemplate.opsForValue().setIfAbsent(key, uuid);
//                    设置过期时间
            redisTemplate.expire(key, 30000, TimeUnit.MILLISECONDS);
            return !StringUtils.isEmpty(result.toString());
        } catch (Exception e) {
            logger.error("set redis occured an exception", e);
        }
        return false;
    }


    /**
     * unlock
     * 每个线程释放锁的时候只能释放自己的锁，即锁必须要有一个拥有者的标记，并且也需要保证释放锁的原子性操作。
     * 在获取锁的时候，可以生成一个随机不唯一的串放入当前线程中，然后再放入 Redis 。
     * 释放锁的时候先判断锁对应的值是否与线程中的值相同，相同时才做删除操作
     * @param key
     * @return
     */
    @Override
    public boolean releaseLock(String key) {
        // 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
        try {
            List<String> keys = new ArrayList<String>();
            keys.add(key);
            List<String> args = new ArrayList<String>();
            args.add(lockFlag.get());

            // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
            // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本

            Long result = (Long) redisTemplate.execute(new RedisCallback<Long>() {
                @Override
                public Long doInRedis(RedisConnection connection) throws DataAccessException {
                    Object nativeConnection = connection.getNativeConnection();
                    // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                    // 集群模式
                    if (nativeConnection instanceof JedisCluster) {
                        return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, args);
                    }

                    // 单机模式
                    else if (nativeConnection instanceof Jedis) {
                        return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, args);
                    }
                    return 0L;
                }
            });

            return result != null && result > 0;
        } catch (Exception e) {
            logger.error("release lock occured an exception", e);
        }
        return false;
    }
}
